package com.mjf.spark.day06

import java.sql.DriverManager

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 从MySQL数据库中读取数据
 */
object Spark04_MySQL_Read {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark04_MySQL_Read")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    // 数据库连接四要素
    var driver = "com.mysql.jdbc.Driver"
    var url = "jdbc:mysql://hadoop103:3306/gmall"
    var username = "root"
    var password = "123456"

    var sql: String = "select * from base_category1 where id > ? and id < ?"

    /**
     * sc: SparkContext,    Spark程序执行入口，上下文对象
     * getConnection: () => Connection,   获取数据库连接
     * sql: String,         执行SQL语句，必须包含两个 ? 占位符
     * lowerBound: Long,    查询的起始位置
     * upperBound: Long,    查询的结束位置
     * numPartitions: Int,  分区个数
     * mapRow: (ResultSet) => T   对结果集处理
     *
     * JDBC操作步骤：
     *    1.注册驱动
     *    2.获取连接
     *    3.创建数据库操作对象PrepareStatement
     *    4.执行SQL
     *    5.处理结果集
     *    6.关闭连接
     */
    val resRDD: JdbcRDD[(Int, String)] = new JdbcRDD(
      sc,
      () => {
        // 注册驱动
        Class.forName(driver)
        // 获取连接
        DriverManager.getConnection(url, username, password)
      },
      // 执行SQL
      sql,
      1,
      20,
      2,
      rs => (rs.getInt(1), rs.getString(2))
    )

    resRDD.collect().foreach(println)

    // 关闭连接
    sc. stop()

  }
}
